use once_cell::sync::Lazy;
use prometheus::{
  register_histogram, register_int_counter, register_int_gauge, Encoder, Histogram, IntCounter,
  IntGauge, TextEncoder,
};

use crate::error::*;
use hyper::header::CONTENT_TYPE;

pub static GAME_SESSIONS: Lazy<IntGauge> =
  Lazy::new(|| register_int_gauge!("flonode_game_sessions", "Number of game sessions").unwrap());
pub static PLAYERS_CONNECTIONS: Lazy<IntGauge> = Lazy::new(|| {
  register_int_gauge!(
    "flonode_player_connections",
    "Number of players connections"
  )
  .unwrap()
});
pub static PLAYER_TOKENS: Lazy<IntGauge> = Lazy::new(|| {
  register_int_gauge!(
    "flonode_player_tokens",
    "Number of registered player tokens"
  )
  .unwrap()
});
pub static ACTIVE_PROXY_CONNECTIONS: Lazy<IntGauge> = Lazy::new(|| {
  register_int_gauge!(
    "flonode_active_proxy_connections",
    "Number of active TCP connections tracked by netfilter on proxy port range"
  )
  .unwrap()
});
pub static PACKET_DROPS: Lazy<IntGauge> = Lazy::new(|| {
  register_int_gauge!(
    "flonode_packet_drops_total",
    "Total number of packets dropped by kernel"
  )
  .unwrap()
});
pub static SOFTNET_BACKLOG_LEN: Lazy<IntGauge> = Lazy::new(|| {
  register_int_gauge!(
    "flonode_softnet_backlog_length",
    "Current softnet backlog queue length"
  )
  .unwrap()
});
pub static NET_RX_PACKETS: Lazy<IntGauge> = Lazy::new(|| {
  register_int_gauge!(
    "flonode_net_rx_packets_total",
    "Total received packets across all interfaces"
  )
  .unwrap()
});
pub static NET_TX_PACKETS: Lazy<IntGauge> = Lazy::new(|| {
  register_int_gauge!(
    "flonode_net_tx_packets_total",
    "Total transmitted packets across all interfaces"
  )
  .unwrap()
});
pub static NET_RX_ERRORS: Lazy<IntGauge> = Lazy::new(|| {
  register_int_gauge!(
    "flonode_net_rx_errors_total",
    "Total receive errors across all interfaces"
  )
  .unwrap()
});
pub static NET_TX_ERRORS: Lazy<IntGauge> = Lazy::new(|| {
  register_int_gauge!(
    "flonode_net_tx_errors_total",
    "Total transmit errors across all interfaces"
  )
  .unwrap()
});
pub static NET_RX_DROPS: Lazy<IntGauge> = Lazy::new(|| {
  register_int_gauge!(
    "flonode_net_rx_drops_total",
    "Total receive drops across all interfaces"
  )
  .unwrap()
});
pub static NET_TX_DROPS: Lazy<IntGauge> = Lazy::new(|| {
  register_int_gauge!(
    "flonode_net_tx_drops_total",
    "Total transmit drops across all interfaces"
  )
  .unwrap()
});
pub static TCP_RETRANS_SEGS: Lazy<IntGauge> = Lazy::new(|| {
  register_int_gauge!(
    "flonode_tcp_retrans_segs_total",
    "Total TCP segments retransmitted"
  )
  .unwrap()
});
pub static UDP_RCV_BUF_ERRORS: Lazy<IntGauge> = Lazy::new(|| {
  register_int_gauge!(
    "flonode_udp_rcv_buf_errors_total",
    "Total UDP receive buffer errors"
  )
  .unwrap()
});
pub static UDP_SND_BUF_ERRORS: Lazy<IntGauge> = Lazy::new(|| {
  register_int_gauge!(
    "flonode_udp_snd_buf_errors_total",
    "Total UDP send buffer errors"
  )
  .unwrap()
});
pub static SOCKETS_USED: Lazy<IntGauge> =
  Lazy::new(|| register_int_gauge!("flonode_sockets_used", "Number of sockets in use").unwrap());
pub static SOFTIRQ_NET_RX: Lazy<IntGauge> = Lazy::new(|| {
  register_int_gauge!("flonode_softirq_net_rx_total", "Total NET_RX softirq count").unwrap()
});
pub static SOFTIRQ_NET_TX: Lazy<IntGauge> = Lazy::new(|| {
  register_int_gauge!("flonode_softirq_net_tx_total", "Total NET_TX softirq count").unwrap()
});


pub static PACKETS_PROCESSED_TOTAL: Lazy<IntCounter> = Lazy::new(|| {
  register_int_counter!(
    "flonode_packets_processed_total",
    "Total number of packets processed"
  )
  .unwrap()
});

fn count_active_proxy_connections() -> Result<i64> {
  use std::process::Command;

  // Use conntrack command to query netfilter connection tracking
  let output = Command::new("conntrack")
    .args(&[
      "-L", // List connections
      "-p", "tcp", // TCP protocol only
    ])
    .output()
    .map_err(|e| {
      tracing::error!("Failed to execute conntrack command: {}", e);
      crate::error::Error::Net(flo_net::error::Error::Io(e))
    })?;

  if !output.status.success() {
    let stderr = String::from_utf8_lossy(&output.stderr);
    tracing::error!(
      "conntrack command failed with status {}: {}",
      output.status,
      stderr
    );
    return Err(crate::error::Error::Net(flo_net::error::Error::Io(
      std::io::Error::new(
        std::io::ErrorKind::Other,
        format!("conntrack failed: {}", stderr),
      ),
    )));
  }

  let stdout = String::from_utf8_lossy(&output.stdout);
  count_conntrack_output(&stdout)
}

fn count_conntrack_output(content: &str) -> Result<i64> {
  let mut count = 0;

  for line in content.lines() {
    if line.trim().is_empty() {
      continue;
    }

    // Skip connections that are not actively transferring data
    if line.contains("TIME_WAIT") || line.contains("CLOSE") {
      continue;
    }

    // conntrack -L output format example:
    // tcp      6 431999 ESTABLISHED src=192.168.1.100 dst=192.168.1.200 sport=12345 dport=80 [ASSURED] mark=0 use=1
    // tcp      6 src=94.139.28.75 dst=45.139.163.21 sport=9606 dport=10113 src=85.90.216.7 dst=45.139.163.21 sport=3554 dport=9606 [OFFLOAD] mark=0 use=2
    let mut dst_port = None;

    // Parse the line to find destination port (local listening port)
    for field in line.split_whitespace() {
      if let Some(port_str) = field.strip_prefix("dport=") {
        if let Ok(port) = port_str.parse::<u16>() {
          dst_port = Some(port);
          break;
        }
      }
    }

    // Count connections where the destination port (local listening port) is in our proxy range
    if let Some(dport) = dst_port {
      if dport >= 10000 && dport <= 50000 {
        count += 1;
      }
    }
  }

  Ok(count)
}

fn read_softnet_stats() -> Result<(i64, i64)> {
  use std::fs;

  let content = fs::read_to_string("/proc/net/softnet_stat").map_err(|e| {
    tracing::warn!("Failed to read /proc/net/softnet_stat: {}", e);
    crate::error::Error::Net(flo_net::error::Error::Io(e))
  })?;

  let mut total_drops = 0i64;
  let mut total_backlog = 0i64;

  for line in content.lines() {
    let fields: Vec<&str> = line.split_whitespace().collect();
    if fields.len() < 3 {
      continue;
    }

    // Fields in /proc/net/softnet_stat:
    // 0: processed packets
    // 1: dropped packets
    // 2: time_squeeze
    // 3: cpu_collision (if present)
    // 4: received_rps (if present)
    // 5: flow_limit_count (if present)
    // 6: softnet_backlog_len (if present)

    if let Ok(drops) = u32::from_str_radix(fields[1], 16) {
      total_drops += drops as i64;
    }

    // Backlog length is typically in field 6 if available
    if fields.len() > 6 {
      if let Ok(backlog) = u32::from_str_radix(fields[6], 16) {
        total_backlog += backlog as i64;
      }
    }
  }

  Ok((total_drops, total_backlog))
}

fn read_net_dev_stats() -> Result<(i64, i64, i64, i64, i64, i64)> {
  use std::fs;

  let content = fs::read_to_string("/proc/net/dev").map_err(|e| {
    tracing::warn!("Failed to read /proc/net/dev: {}", e);
    crate::error::Error::Net(flo_net::error::Error::Io(e))
  })?;

  let mut rx_packets = 0i64;
  let mut tx_packets = 0i64;
  let mut rx_errors = 0i64;
  let mut tx_errors = 0i64;
  let mut rx_drops = 0i64;
  let mut tx_drops = 0i64;

  for line in content.lines().skip(2) {
    let line = line.trim();
    if line.is_empty() {
      continue;
    }

    let parts: Vec<&str> = line.split_whitespace().collect();
    if parts.len() < 17 {
      continue;
    }

    let interface = parts[0].trim_end_matches(':');
    if interface == "lo" {
      continue;
    }

    if let (Ok(rx_pkt), Ok(tx_pkt), Ok(rx_err), Ok(tx_err), Ok(rx_drop), Ok(tx_drop)) = (
      parts[2].parse::<i64>(),
      parts[10].parse::<i64>(),
      parts[3].parse::<i64>(),
      parts[11].parse::<i64>(),
      parts[4].parse::<i64>(),
      parts[12].parse::<i64>(),
    ) {
      rx_packets += rx_pkt;
      tx_packets += tx_pkt;
      rx_errors += rx_err;
      tx_errors += tx_err;
      rx_drops += rx_drop;
      tx_drops += tx_drop;
    }
  }

  Ok((
    rx_packets, tx_packets, rx_errors, tx_errors, rx_drops, tx_drops,
  ))
}

fn read_tcp_retrans_stats() -> Result<i64> {
  use std::fs;

  let content = fs::read_to_string("/proc/net/netstat").map_err(|e| {
    tracing::warn!("Failed to read /proc/net/netstat: {}", e);
    crate::error::Error::Net(flo_net::error::Error::Io(e))
  })?;

  for line in content.lines() {
    if line.starts_with("TcpExt:") && line.contains("RetransSegs") {
      let header_line = line;
      if let Some(next_line) = content
        .lines()
        .find(|l| l.starts_with("TcpExt:") && l != &header_line)
      {
        let headers: Vec<&str> = header_line.split_whitespace().collect();
        let values: Vec<&str> = next_line.split_whitespace().collect();

        for (i, header) in headers.iter().enumerate() {
          if *header == "RetransSegs" && i < values.len() {
            if let Ok(retrans) = values[i].parse::<i64>() {
              return Ok(retrans);
            }
          }
        }
      }
    }
  }

  Ok(0)
}

fn read_udp_buffer_errors() -> Result<(i64, i64)> {
  use std::fs;

  let content = fs::read_to_string("/proc/net/snmp").map_err(|e| {
    tracing::warn!("Failed to read /proc/net/snmp: {}", e);
    crate::error::Error::Net(flo_net::error::Error::Io(e))
  })?;

  let mut rcv_buf_errors = 0i64;
  let mut snd_buf_errors = 0i64;

  for line in content.lines() {
    if line.starts_with("Udp:") && line.contains("RcvbufErrors") {
      let header_line = line;
      if let Some(next_line) = content
        .lines()
        .find(|l| l.starts_with("Udp:") && l != &header_line)
      {
        let headers: Vec<&str> = header_line.split_whitespace().collect();
        let values: Vec<&str> = next_line.split_whitespace().collect();

        for (i, header) in headers.iter().enumerate() {
          if i < values.len() {
            match *header {
              "RcvbufErrors" => {
                if let Ok(errors) = values[i].parse::<i64>() {
                  rcv_buf_errors = errors;
                }
              }
              "SndbufErrors" => {
                if let Ok(errors) = values[i].parse::<i64>() {
                  snd_buf_errors = errors;
                }
              }
              _ => {}
            }
          }
        }
      }
    }
  }

  Ok((rcv_buf_errors, snd_buf_errors))
}

fn read_socket_stats() -> Result<i64> {
  use std::fs;

  let content = fs::read_to_string("/proc/net/sockstat").map_err(|e| {
    tracing::warn!("Failed to read /proc/net/sockstat: {}", e);
    crate::error::Error::Net(flo_net::error::Error::Io(e))
  })?;

  for line in content.lines() {
    if line.starts_with("sockets: used") {
      let parts: Vec<&str> = line.split_whitespace().collect();
      if parts.len() >= 3 {
        if let Ok(used) = parts[2].parse::<i64>() {
          return Ok(used);
        }
      }
    }
  }

  Ok(0)
}

fn read_softirq_stats() -> Result<(i64, i64)> {
  use std::fs;

  let content = fs::read_to_string("/proc/softirqs").map_err(|e| {
    tracing::warn!("Failed to read /proc/softirqs: {}", e);
    crate::error::Error::Net(flo_net::error::Error::Io(e))
  })?;

  let mut net_rx_total = 0i64;
  let mut net_tx_total = 0i64;

  for line in content.lines() {
    let parts: Vec<&str> = line.split_whitespace().collect();
    if parts.is_empty() {
      continue;
    }

    let irq_type = parts[0].trim_end_matches(':');
    match irq_type {
      "NET_RX" => {
        for i in 1..parts.len() {
          if let Ok(count) = parts[i].parse::<i64>() {
            net_rx_total += count;
          }
        }
      }
      "NET_TX" => {
        for i in 1..parts.len() {
          if let Ok(count) = parts[i].parse::<i64>() {
            net_tx_total += count;
          }
        }
      }
      _ => {}
    }
  }

  Ok((net_rx_total, net_tx_total))
}

async fn update_proxy_connections_metric() {
  let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(15));

  loop {
    interval.tick().await;

    match count_active_proxy_connections() {
      Ok(count) => {
        ACTIVE_PROXY_CONNECTIONS.set(count);
      }
      Err(e) => {
        tracing::warn!("Failed to update proxy connections metric: {}", e);
      }
    }
  }
}

async fn update_network_metrics() {
  let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(15));

  loop {
    interval.tick().await;

    match read_softnet_stats() {
      Ok((drops, backlog)) => {
        PACKET_DROPS.set(drops);
        SOFTNET_BACKLOG_LEN.set(backlog);
      }
      Err(e) => {
        tracing::warn!("Failed to update softnet metrics: {}", e);
      }
    }

    match read_net_dev_stats() {
      Ok((rx_packets, tx_packets, rx_errors, tx_errors, rx_drops, tx_drops)) => {
        NET_RX_PACKETS.set(rx_packets);
        NET_TX_PACKETS.set(tx_packets);
        NET_RX_ERRORS.set(rx_errors);
        NET_TX_ERRORS.set(tx_errors);
        NET_RX_DROPS.set(rx_drops);
        NET_TX_DROPS.set(tx_drops);
      }
      Err(e) => {
        tracing::warn!("Failed to update network interface metrics: {}", e);
      }
    }

    match read_tcp_retrans_stats() {
      Ok(retrans) => {
        TCP_RETRANS_SEGS.set(retrans);
      }
      Err(e) => {
        tracing::warn!("Failed to update TCP retransmission metrics: {}", e);
      }
    }

    match read_udp_buffer_errors() {
      Ok((rcv_errors, snd_errors)) => {
        UDP_RCV_BUF_ERRORS.set(rcv_errors);
        UDP_SND_BUF_ERRORS.set(snd_errors);
      }
      Err(e) => {
        tracing::warn!("Failed to update UDP buffer error metrics: {}", e);
      }
    }

    match read_socket_stats() {
      Ok(sockets_used) => {
        SOCKETS_USED.set(sockets_used);
      }
      Err(e) => {
        tracing::warn!("Failed to update socket metrics: {}", e);
      }
    }

    match read_softirq_stats() {
      Ok((net_rx, net_tx)) => {
        SOFTIRQ_NET_RX.set(net_rx);
        SOFTIRQ_NET_TX.set(net_tx);
      }
      Err(e) => {
        tracing::warn!("Failed to update softirq metrics: {}", e);
      }
    }
  }
}

pub async fn serve_metrics() -> Result<()> {
  use bytes::Bytes;
  use http_body_util::Full;
  use hyper::{body, service::service_fn, Request, Response};
  use hyper_util::rt::TokioExecutor;
  use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};

  async fn serve_req(req: Request<body::Incoming>) -> Result<Response<Full<Bytes>>, hyper::Error> {
    if req.uri().path() == "/version" {
      let response = Response::builder()
        .status(200)
        .body(Full::new(crate::version::FLO_NODE_VERSION_STRING.into()))
        .unwrap();

      return Ok(response);
    }

    let encoder = TextEncoder::new();

    let metric_families = prometheus::gather();
    let mut buffer = vec![];
    encoder.encode(&metric_families, &mut buffer).unwrap();

    let response = Response::builder()
      .status(200)
      .header(CONTENT_TYPE, encoder.format_type())
      .body(Full::new(buffer.into()))
      .unwrap();

    Ok(response)
  }

  let addr = SocketAddr::from(SocketAddrV4::new(
    Ipv4Addr::UNSPECIFIED,
    flo_constants::NODE_HTTP_PORT,
  ));

  let listener = tokio::net::TcpListener::bind(&addr).await?;

  tokio::spawn(update_proxy_connections_metric());
  tokio::spawn(update_network_metrics());

  loop {
    let (stream, _) = listener.accept().await?;

    let io = hyper_util::rt::TokioIo::new(stream);

    tokio::spawn(async move {
      // use `auto::Builder` is for supporting both HTTP/1 and HTTP/2 at the same time.
      if let Err(err) = hyper_util::server::conn::auto::Builder::new(TokioExecutor::new())
        .serve_connection(io, service_fn(serve_req))
        .await
      {
        tracing::error!("Error serving connection: {}", err);
      }
    });
  }
}
